package com.dsp.kafka.demo.rest;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaProducerException;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;

@RestController
public class KafkaProducer {
    Logger logger = LoggerFactory.getLogger(KafkaProducer.class);

    @Resource
    private KafkaTemplate<String,String> template;

    /**
     * test:
     * http://localhost:8080/send?topic=OneQueue&message=nihao
     */
    @RequestMapping("/send")
    public String sendMsg(String topic, String message) throws ExecutionException, InterruptedException {
        // List<ListenableFuture<SendResult<String, String>>> futures = new ArrayList<>();
        // CountDownLatch latch = new CountDownLatch(toSend.size());

        ListenableFuture<SendResult<String,String>> listenableFuture = template.send(topic, message);
        ListenableFutureCallback<SendResult<String, String>> callback =
                new ListenableFutureCallback<SendResult<String, String>>() {

                    @Override
                    public void onSuccess(SendResult<String, String> result) {
                        assert result != null;
                        logger.info(result.getRecordMetadata().toString());
                        // latch.countDown();
                    }

                    @Override
                    public void onFailure(Throwable ex) {
                        ProducerRecord<String, String> producerRecord = ((KafkaProducerException) ex).getFailedProducerRecord();
                        logger.error("Failed; " + producerRecord, ex);
                        // latch.countDown();
                    }
                };
        listenableFuture.addCallback(callback);
        return "success";
    }
}
